package tech.pdai.springboot.redis.jedis.service;

import com.alibaba.fastjson2.JSON;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;

import javax.annotation.Resource;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 基于Redis List的消息队列
 */
@Slf4j
@Service
public class RedisMessageQueue {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    /**
     * 消息队列键前缀
     */
    private static final String QUEUE_PREFIX = "queue:";
    /**
     * 死信队列键前缀
     */
    private static final String DEAD_LETTER_QUEUE_PREFIX = "queue:dead:";
    /**
     * 阻塞超时时间(秒)
     */
    private static final int BLOCK_TIMEOUT = 30;

    /**
     * 发送消息
     * *@paramqueueName 队列名称
     * *@parammessage 消息内容
     * *@return发送成功后的队列长度
     */
    public Long sendMessage(String queueName, Object message) {
        Assert.notNull(queueName, "队列名称不能为空");
        if (message == null) {
            throw new IllegalArgumentException("消息内容不能为空");
        }
        String key = QUEUE_PREFIX + queueName;
        String messageJson = JSON.toJSONString(message);
        // 从左侧插入消息
        Long size = redisTemplate.opsForList().leftPush(key, messageJson);
        log.info("发送消息到队列，queueName:{}, 消息内容:{}, 队列长度:{}", queueName, messageJson, size);
        return size;
    }

    /**
     * 接收消息（非阻塞）  *  *@paramqueueName 队列名称  *@paramclazz 消息类型  *@return消息对象，null表示没有消息
     */
    public <T> T receiveMessage(String queueName, Class<T> clazz) {
        Assert.notNull(queueName, "队列名称不能为空");
        String key = QUEUE_PREFIX + queueName;
        // 从右侧弹出消息
        Object messageJson = redisTemplate.opsForList().rightPop(key);
        if (Objects.isNull(messageJson)) {
            log.debug("队列中没有消息，queueName:{}", queueName);
            return null;
        }
        log.info("从队列接收消息，queueName:{}, 消息内容:{}", queueName, messageJson);
        return JSON.parseObject(messageJson.toString(), clazz);
    }

    /**
     * 接收消息（阻塞）  *  *@paramqueueName 队列名称  *@paramclazz 消息类型  *@return消息对象
     */
    public <T> T blockingReceiveMessage(String queueName, Class<T> clazz) {
        return blockingReceiveMessage(queueName, clazz, BLOCK_TIMEOUT);
    }

    /**
     * 接收消息（阻塞，指定超时时间）  *  *@paramqueueName 队列名称  *@paramclazz 消息类型  *@paramtimeout 超时时间(秒)  *@return消息对象，超时返回null
     */
    public <T> T blockingReceiveMessage(String queueName, Class<T> clazz, int timeout) {
        Assert.notNull(queueName, "队列名称不能为空");
        if (timeout <= 0) {
            throw new IllegalArgumentException("超时时间必须大于0");
        }
        String key = QUEUE_PREFIX + queueName;
        // 阻塞式从右侧弹出消息
        Object result = redisTemplate.opsForList().rightPop(key, timeout, TimeUnit.SECONDS);
        if (result == null) {
            log.debug("队列接收消息超时，queueName:{}, timeout:{}s", queueName, timeout);
            return null;
        }
        String messageJson = result.toString();
        log.info("从队列阻塞接收消息，queueName:{}, 消息内容:{}", queueName, messageJson);
        return JSON.parseObject(messageJson, clazz);
    }

    /**
     * 将消息放入死信队列  *  *@paramqueueName 原队列名称  *@parammessage 消息内容  *@paramreason 失败原因
     */
    public void sendToDeadLetterQueue(String queueName, Object message, String reason) {
        Assert.notNull(queueName, "队列名称不能为空");
        if (message == null) {
            throw new IllegalArgumentException("消息内容不能为空");
        }
        String key = DEAD_LETTER_QUEUE_PREFIX + queueName;
        // 构建死信消息
        DeadLetterMessage deadLetter = new DeadLetterMessage();
        deadLetter.setMessage(message);
        deadLetter.setReason(reason);
        deadLetter.setTimestamp(System.currentTimeMillis());
        String messageJson = JSON.toJSONString(deadLetter);
        // 放入死信队列
        redisTemplate.opsForList().leftPush(key, messageJson);
        log.warn("消息放入死信队列，queueName:{}, reason:{}, 消息内容:{}", queueName, reason, messageJson);
    }

    /**
     * 获取队列长度  *  *@paramqueueName 队列名称  *@return队列长度
     */
    public Long getQueueSize(String queueName) {
        Assert.notNull(queueName, "队列名称不能为空");
        String key = QUEUE_PREFIX + queueName;
        return redisTemplate.opsForList().size(key);
    }

    /**
     * 死信消息封装类
     */
    @Getter
    @Setter
    public static class DeadLetterMessage {
        private Object message;
        private String reason;
        private long timestamp;
        // getter和setter省略
    }
}
